package com.hlz.flink.chapter05;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * reduce操作
 *
 * @author Hongliang Zhu
 * @create 2023-02-16 23:29
 */
public class TransReduceTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

//        DataStreamSource<Event> source = env.addSource(new ClickSource());

        DataStreamSource<Event> source = env.fromElements(
                new Event("Mary", "./mary", 1000L),
                new Event("Bob", "./hello", 2000L),
                new Event("Alice", "./prod", 3000L),
                new Event("Alice", "./prod2", 4000L),
                new Event("Bob", "./prod", 5000L),
                new Event("Bob", "./prod1", 6000L),
                new Event("Bob", "./prod2", 7000L),
                new Event("Bob", "./prod3", 8000L),
                new Event("Alice", "./prod2", 4000L),
                new Event("Alice", "./prod2", 4000L),
                new Event("Alice", "./prod2", 4000L),
        new Event("Alice", "./prod2", 4000L),
                new Event("Alice", "./prod2", 4000L));


        // 将Event类型转换成Tuple元组类型
//        SingleOutputStreamOperator<Tuple3<String, String, Long>> tupleStream = source.map(e -> Tuple3.of(e.getUser(), e.getUrl(), 1L)).returns(new TypeHint<Tuple3<String, String, Long>>() {
//        });

        SingleOutputStreamOperator<Tuple3<String, String, Long>> reduceStream = source
                .map(e -> Tuple3.of(e.getUser(), e.getUrl(), 1L))
                .returns(new TypeHint<Tuple3<String, String, Long>>() {
                })
                .keyBy(t -> t.f0)
                .reduce(new ReduceFunction<Tuple3<String, String, Long>>() {
                    @Override
                    public Tuple3<String, String, Long> reduce(Tuple3<String, String, Long> value1, Tuple3<String, String, Long> value2) throws Exception {

                        // 每到一条数据，pv加1
                        return Tuple3.of(value1.f0, value2.f1, value1.f2 + value2.f2);
                    }
                }).keyBy(r -> "hello").reduce(new ReduceFunction<Tuple3<String, String, Long>>() {

                    // 找最大值
                    @Override
                    public Tuple3<String, String, Long> reduce(Tuple3<String, String, Long> value1, Tuple3<String, String, Long> value2) throws Exception {
                        return value1.f2 > value2.f2 ? value1 : value2;
                    }
                });

        reduceStream.print("reduceStream");

        env.execute();
    }
}
